package org.jie.guider.framework.springboot.kafka;

import java.util.ArrayList;
import java.util.Collection;
import java.util.regex.Pattern;
import org.springframework.kafka.config.KafkaListenerEndpoint;
import org.springframework.kafka.listener.BatchMessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.kafka.support.converter.MessageConverter;

/**
 * @author LiJie
 */
public class ChannelKafkaListenerEndpoint implements KafkaListenerEndpoint {
    /**
     * 通道 ID
     */
    private Long channel;
    /**
     * 主题
     */
    private String topic;
    /**
     * 端点 ID
     */
    private String id;
    /**
     * 消费组
     */
    private String group;
    /**
     * 并发数
     */
    private Integer concurrency;

    public ChannelKafkaListenerEndpoint(Long channel, Integer concurrency) {
        this.channel = channel;
        this.topic = "channel_" + channel;
        this.concurrency = concurrency;
        this.id = "channel_" + channel;
        this.group = "channel_group_" + channel;
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public String getGroupId() {
        return group;
    }

    @Override
    public String getGroup() {
        return group;
    }

    @Override
    public Collection<String> getTopics() {
        Collection<String> topics = new ArrayList<>();
        topics.add(topic);
        return topics;
    }

    @Override
    public TopicPartitionOffset[] getTopicPartitionsToAssign() {
        return new TopicPartitionOffset[0];
    }

    @Override
    public Pattern getTopicPattern() {
        return null;
    }

    @Override
    public String getClientIdPrefix() {
        return null;
    }

    @Override
    public Integer getConcurrency() {
        return concurrency;
    }

    @Override
    public Boolean getAutoStartup() {
        return null;
    }

    @Override
    public void setupListenerContainer(MessageListenerContainer listenerContainer,
                                       MessageConverter messageConverter) {
        BatchMessageListener<Integer, String> listener = new ChannelMessageListener<>(channel);
        listenerContainer.setupMessageListener(listener);
    }

    @Override
    public boolean isSplitIterables() {
        return false;
    }
}
